package com.xzj.common.plugins.disruptor.mq;

import com.alibaba.fastjson2.JSON;
import com.lmax.disruptor.EventHandler;
import com.xzj.common.plugins.disruptor.domain.DisruptorMessage;
import lombok.extern.slf4j.Slf4j;

import java.util.function.Consumer;

/**
 * @program: xzj-common
 * @ClassName: DisruptorConsumer
 * @description: 消息消费者
 * @author: XiaZunJun
 * @create: 2024-11-19 10:36
 */
@Slf4j
public class DisruptorConsumer<T> implements EventHandler<DisruptorMessage<T>> {

    private final Consumer<T> consumer;

    public DisruptorConsumer(Consumer<T> consumer) {
        this.consumer = consumer;
    }

    @Override
    public void onEvent(DisruptorMessage<T> messageModel, long l, boolean b) {
        log.debug("消费者处理消息开始");
        try {
            if (messageModel != null) {
                log.debug("消费者消费的信息是：{}", JSON.toJSONString(messageModel));
                consumer.accept(messageModel.getT());
            }
        } catch (Exception e) {
            log.error("消费者处理消息失败",e);
        }
        log.debug("消费者处理消息结束");
    }
}